use std::{
    sync::{
        atomic::{AtomicU8, Ordering},
        Arc,
    },
    time::Duration,
};

use super::frame::{
    make, make_empty, RecvFrame, TYPE_ACK, TYPE_NOTIFY, TYPE_NOTIFY_ACK, TYPE_PING, TYPE_PONG, TYPE_REQ, TYPE_RES,
};
use super::model::LoginRes;
use super::reader::read_frame;
use super::service::wait_login;
use dashmap::DashMap;
use jni::objects::{GlobalRef, JValue};
use log::debug;
use tokio::{
    io::{split, AsyncWriteExt, BufReader, ReadHalf, WriteHalf},
    net::TcpStream,
    sync::{oneshot, Mutex, Semaphore},
    time::timeout,
};

use crate::{
    error::{new_err, AppErr, IoErr},
    utils::global::get_java_vm,
};

pub type RespMap = DashMap<u16, oneshot::Sender<RecvFrame>>;
pub type Reader = BufReader<ReadHalf<TcpStream>>;

pub struct Conn {
    writer: Mutex<WriteHalf<TcpStream>>,
    resp_map: RespMap,
    exit_sem: Semaphore,
    seq: AtomicU8,
    cb: GlobalRef,
    pub info: LoginRes,
}

pub type SharedConn = Arc<Conn>;

pub async fn new_conn(addr: &str, cb: GlobalRef) -> Result<SharedConn, AppErr> {
    let stream = timeout(Duration::from_secs(10), TcpStream::connect(addr))
        .await
        .map_err(|_| new_err("连接服务器超时"))??;

    let (reader, mut writer) = split(stream);
    let mut reader = BufReader::new(reader);

    let res = wait_login(&mut writer, &mut reader).await?;

    let conn = Conn {
        writer: Mutex::new(writer),
        resp_map: RespMap::new(),
        exit_sem: Semaphore::new(0),
        seq: AtomicU8::new(0),
        cb,
        info: res,
    };
    let conn = Arc::new(conn);
    conn.on_login()?;
    tokio::spawn(read_task(conn.clone(), reader));

    Ok(conn)
}

async fn read_loop(conn: SharedConn, reader: &mut Reader) -> Result<(), AppErr> {
    loop {
        let frame = tokio::select! {
            _ = conn.exit_sem.acquire() => break,
            frame = read_frame(reader) => frame?
        };

        match frame.ctype() {
            TYPE_PING => {
                conn.pong(frame.seq()).await?;
            }

            TYPE_PONG | TYPE_ACK => {
                conn.on_resp(frame);
            }

            TYPE_NOTIFY => conn.on_recv("onNotify", "(I[B)V", frame)?,

            TYPE_NOTIFY_ACK => {
                conn.ack(frame.seq()).await?;
                conn.on_recv("onNotifyAck", "(I[B)V", frame)?;
            }

            TYPE_REQ => {
                conn.on_recv("onReq", "(I[B)V", frame)?;
            }

            _ => {}
        }
    }
    Ok(())
}

async fn read_task(conn: SharedConn, mut reader: Reader) {
    let r = read_loop(conn, &mut reader).await;
    if let Err(e) = r {
        debug!("read err {}", e);
    }
    debug!("read task exit");
}

impl Conn {
    pub async fn req(&self, cmd: u8, args: &[u8]) -> Result<RecvFrame, AppErr> {
        let seq = self.create_seq();
        let buf = make(TYPE_REQ, seq, cmd, args);
        let frame = self
            .exec(TYPE_RES, seq, &buf, Duration::from_secs(10))
            .await?;
        Ok(frame)
    }

    pub async fn ping(&self) -> Result<(), AppErr> {
        let seq = self.create_seq();
        let buf = make_empty(TYPE_PING, seq);
        self.exec(TYPE_PONG, seq, &buf, Duration::from_secs(2))
            .await?;
        Ok(())
    }

    pub async fn notify(&self, cmd: u8, args: &[u8]) -> Result<(), AppErr> {
        let buf = make(TYPE_NOTIFY, 0, cmd, args);
        self.write(&buf).await?;
        Ok(())
    }

    pub async fn notify_ack(&self, cmd: u8, args: &[u8]) -> Result<(), AppErr> {
        let seq = self.create_seq();
        let buf = make(TYPE_NOTIFY_ACK, seq, cmd, args);
        self.exec(TYPE_ACK, seq, &buf, Duration::from_secs(2))
            .await?;
        Ok(())
    }

    pub fn close(&self) {
        self.exit_sem.add_permits(2);
    }

    async fn exec(
        &self,
        r_ctype: u8,
        seq: u8,
        buf: &[u8],
        duration: Duration,
    ) -> Result<RecvFrame, AppErr> {
        let rx = self.create_resp(r_ctype, seq);
        self.write(buf).await?;
        let frame = timeout(duration, rx)
            .await
            .map_err(|_| new_err("返回超时"))?
            .map_err(|_| new_err("recv err"))?;
        Ok(frame)
    }

    fn create_seq(&self) -> u8 {
        self.seq.fetch_add(1, Ordering::Relaxed)
    }

    fn create_resp(&self, r_ctype: u8, seq: u8) -> oneshot::Receiver<RecvFrame> {
        let id = ((r_ctype as u16) << 8) + (seq as u16);
        let (tx, rx) = oneshot::channel();
        self.resp_map.insert(id, tx);
        rx
    }

    fn on_resp(&self, frame: RecvFrame) {
        let id = frame.id();
        let ret = self.resp_map.remove(&id);
        if let Some((_, v)) = ret {
            _ = v.send(frame);
        }
    }

    fn on_recv(&self, name: &str, sig: &str, frame: RecvFrame) -> Result<(), AppErr> {
        let mut env = get_java_vm().attach_current_thread_permanently()?;
        let body = env.byte_array_from_slice(frame.body()?)?;
        env.call_method(
            &self.cb,
            name,
            sig,
            &[JValue::Int(frame.cmd()? as i32), JValue::Object(&body)],
        )?;

        Ok(())
    }

    async fn pong(&self, seq: u8) -> Result<(), IoErr> {
        let buf = make_empty(TYPE_PONG, seq);
        self.write(&buf).await?;
        Ok(())
    }

    async fn ack(&self, seq: u8) -> Result<(), AppErr> {
        let buf = make_empty(TYPE_ACK, seq);
        self.write(&buf).await?;
        Ok(())
    }

    async fn write(&self, buf: &[u8]) -> Result<(), IoErr> {
        self.writer.lock().await.write_all(buf).await?;
        Ok(())
    }

    fn on_login(&self) -> Result<(), AppErr> {

        let obj = self.info.to_java()?;
        let mut env = get_java_vm().attach_current_thread_permanently()?;

        env.call_method(
            &self.cb, 
            "onLogin", 
        "(Lrustlib/service/LoginInfo;)V", 
        &[JValue::Object(&obj)]
        )?;

        Ok(())
    }
}
